package com.bjsdzk.collector.one2one;

import com.bjsdzk.collector.constant.Constant;
import com.bjsdzk.collector.entity.CommandQueueEle;
import com.bjsdzk.collector.entity.Message;
import com.bjsdzk.collector.utils.TimeUtil;
import com.bjsdzk.collector.utils.Utils;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Handler;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetServerOptions;
import io.vertx.core.net.NetSocket;
import io.vertx.core.shareddata.Counter;
import io.vertx.core.shareddata.SharedData;
import io.vertx.ext.mongo.BulkOperation;
import io.vertx.ext.mongo.MongoClient;
import io.vertx.ext.mongo.UpdateOptions;
import io.vertx.redis.RedisClient;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.math.BigDecimal;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author pcgudunimasigaga
 */
public class OnlyServer extends AbstractVerticle {

    private static Logger logger = LoggerFactory.getLogger(OnlyServer.class);
    public static final  String MONGONAME="RealTimeData";

    RedisClient redis;
    MongoClient mongoClient;

    public OnlyServer() {
    }

    public OnlyServer(RedisClient redis, MongoClient mongoClient) {
        this.redis = redis;
        this.mongoClient = mongoClient;
    }

    EventBus eventBus = null;

    Integer port;
    Integer dtuType;
    String  deviceType;
    String  serverAddress;
    Map<String, Map<String,Integer>> electricParmas = new HashMap<>();

    Map<String, Integer> handlerIdMap = new ConcurrentHashMap<>(256);
    Map<String, Integer> lineStatusMap = new ConcurrentHashMap<>(256);
    Map<String, List<Long>> timerOther = new ConcurrentHashMap<>(256);

    AtomicInteger conectCount = new AtomicInteger(0);

    @Override
    public void start() {
        try {
            eventBus = vertx.eventBus();
            port = Integer.parseInt(config().getString("port"));
            dtuType = Integer.parseInt(config().getString("dtuType"));
            deviceType = config().getString("deviceType");

            if (dtuType == 7) {
                serverAddress=config().getString("serverAddress");
                electricParmas=new JsonObject(config().getString("electricParmas")).mapTo(electricParmas.getClass());
                electricParmas.forEach((k,v) ->{
                    lineStatusMap.put(k, 0);
                });

                eventBus.localConsumer(Constant.MOD_SEND_SET_MODEL + port, res -> {
                    JsonObject body = (JsonObject) res.body();
                    electricParmas=new JsonObject(body.getString("electricParmas")).mapTo(electricParmas.getClass());
                    electricParmas.forEach((k,v) ->{
                        if(!lineStatusMap.containsKey(k)){
                            lineStatusMap.put(k, 0);
                        }
                    });
                });
            }



            NetServerOptions options = new NetServerOptions().setTcpNoDelay(true)
                    .setSoLinger(100).setPort(port);
            NetServer server = vertx.createNetServer(options);

            SharedData sharedData = vertx.sharedData();

            server.connectHandler(netSocket -> {

                calcCount(sharedData, true);
                conectCount.incrementAndGet();


                List<String> deviceCodeList = new ArrayList<>();

                PriorityBlockingQueue<Message> queue = new PriorityBlockingQueue<>();
                ConcurrentLinkedQueue<CommandQueueEle> returnQueue = new ConcurrentLinkedQueue<>();

                String handlerID = netSocket.writeHandlerID();
                Map<String, Long> keepLiveMap = new HashMap<>();
                keepLiveMap.put("alive", System.currentTimeMillis());
                timerOther.put(handlerID, new ArrayList<>());

                ShuanghuaHnadler handler = new ShuanghuaHnadler(netSocket, keepLiveMap, queue, returnQueue, serverAddress, deviceCodeList);
                netSocket.handler(handler);

                int interval=1000*60*5;
                int everyTime=1000*60*20;
                if (port==62001){
                    interval=1000*60;
                    everyTime=1000*60;
                }
                checkLine(interval, everyTime, keepLiveMap, netSocket, timerOther.get(handlerID), deviceCodeList);
                startConsumer(netSocket, timerOther.get(handlerID), queue, returnQueue, lineStatusMap, deviceCodeList);

                netSocket.closeHandler(h -> {
                    logger.info("connect close==>port:{}==>HandlerId:{}", port, handlerID);
                    if (!deviceCodeList.isEmpty()) {
                        downLine(deviceCodeList, handlerID);
                        logger.info("connect close==>port:{}==>deviceCode:{}==>HandlerId:{}", port, deviceCodeList.get(0), handlerID);
                    }
                    cancelTimer(timerOther.get(handlerID));
                    calcCount(sharedData, false);
                });

                netSocket.exceptionHandler(e -> {
                    if (deviceCodeList.isEmpty()) {
                        logger.info("-----------------------异常关闭下线 " + "==>port:{}==>HandlerId:{}==>{}", port, handlerID, e.getMessage());
                    } else {
                        logger.info("-----------------------异常关闭下线 " + "==>port:{}==>deviceCode:{}==>HandlerId:{}==>{}", port, deviceCodeList.get(0), handlerID, e.getMessage());
                    }
                    netSocket.close();
                });

            }).listen(port, netServerAsyncResult -> {
                if (netServerAsyncResult.succeeded()) {

                } else {
                    logger.error("start error port:" + port + netServerAsyncResult.cause().getMessage());
                    vertx.undeploy(deploymentID());
                    eventBus.send(Constant.MOD_START_FAIL, config());
                }

            });

        } catch (Exception e) {
            ByteArrayOutputStream stream = new ByteArrayOutputStream();
            e.printStackTrace(new PrintStream(stream));
            String exception = stream.toString();
            logger.error(port+"=exception="+exception);
        }
    }
    class ShuanghuaHnadler implements Handler<Buffer> {
        Map<String, Long> keepLiveMap;
        NetSocket netSocket;
        PriorityBlockingQueue<Message> queue;
        String deviceCode ;
        List<String> deviceCodeList;
        String serverAddress ;
        ConcurrentLinkedQueue<CommandQueueEle> returnQueue;

        public ShuanghuaHnadler(NetSocket netSocket, Map<String, Long> keepLiveMap, PriorityBlockingQueue<Message> queue,ConcurrentLinkedQueue<CommandQueueEle> returnQueue,String serverAddress,List<String> deviceCodeList) {
            this.keepLiveMap = keepLiveMap;
            this.netSocket = netSocket;
            this.queue = queue;
            this.returnQueue = returnQueue;
            this.serverAddress = serverAddress;
            this.deviceCodeList = deviceCodeList;

            setTimer(netSocket);
        }

        private Buffer data = Buffer.buffer();

        @Override
        public void handle(Buffer buffer) {
            try {
                byte[] bytes = data.appendBuffer(buffer).getBytes();
                String data2String = Utils.bytes2String(bytes);
                String handlerId=netSocket.writeHandlerID();
                logger.info("receive data ==>port:{}==>HandlerID:{}==>deviceCode:{}==>" + data2String,port,handlerId,deviceCode);
                data = Buffer.buffer();
                int lengthPosition = 0;
                int other = 20;
                int totalLength;

                if (data2String.startsWith("4745")) {
                    logger.debug("deviceCode:{}==>{}",deviceCode,data2String);
                    return ;
                }

                while (true) {
                    int dataLenth = bytes.length;
                    if (dataLenth > 12&&dataLenth<300) {
                        lengthPosition = Utils.bytestoint(bytes, 12,2);
                    }else if (dataLenth>300){
                        data = Buffer.buffer();
                        break;
                    }else {
                        data = Buffer.buffer(bytes);
                        break;
                    }
                    totalLength = lengthPosition + other;

                    if (dataLenth < totalLength) {
                        data = Buffer.buffer(bytes);
                        logger.info("data lenth not enough");
                        return;
                    } else if (dataLenth == totalLength) {
                        data(bytes,handlerId);
                        break;
                    } else {
                        if (data(Utils.getbytes(bytes, 0, totalLength),handlerId)) {
                            bytes= Utils.getbytes(bytes, totalLength, dataLenth-totalLength);
                        } else {
                            data = Buffer.buffer();
                            break;
                        }
                    }

                }

            } catch (Exception e) {
                ByteArrayOutputStream stream = new ByteArrayOutputStream();
                e.printStackTrace(new PrintStream(stream));
                String exception = stream.toString();
                logger.error(port+deviceCode + "=exception=" + exception);
            }

        }

        private boolean data(byte[] bytes,String handlerId) {
//                终端发送：
//                88 81 00 03 01 00 00 00 00 00 00 00 00 05 04 bb 00 02 33 82 7a 0d 0d 0d 0d
//                平台发送：
//                88 81 00 03 01 00 00 00 00 00 00 00 00 07 01 de f0 e4 59 27 0f 30 da 0d 0d 0d 0d
//                终端解析出地址Get IP (222,240,228,89) 9999
            String data2String = Utils.bytes2String(bytes);
//            logger.debug("parse data ==>port:{}==>HandlerID:{}==>deviceCode:{}==>{}",port,handlerId,deviceCode,data2String);
            String flag = Utils.bytes2String(bytes, 2);
            int datalenth = data2String.length();
            int datalenthNode = datalenth - 12;
            String starcrc = data2String.substring(0, datalenthNode);
            String endcrc = data2String.substring(datalenthNode, datalenth - 8);

            if (!StringUtils.equals(Utils.checkNumSecond(starcrc), endcrc)) {
                logger.error("crc  check no pass ");
                return false;
            } else {
                if (deviceCode!=null){
                    upLine( deviceCode, keepLiveMap,handlerId);
                }
                keepLiveMap.put("alive", System.currentTimeMillis());
            }
            DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
            String currentTime = dateFormat.format(new Date());
            int start = 14;
            switch (flag) {
                case "00":
                    int deviceLenth = Utils.bytestoint(bytes, start);
                    String deviceC = Utils.bytes2String(bytes, start + 1, deviceLenth);
                    String cmd;
//                    if (electricParmas.containsKey(deviceC)) {
                        String[] ad = serverAddress.split("-");
                        String content = "01";
                        for (int i = 0; i < ad.length; i++) {
                            if (i == ad.length - 1) {
                                content = content + String.format("%04x", Integer.parseInt(ad[i]));
                            } else {
                                content = content + String.format("%02x", Integer.parseInt(ad[i]));

                            }
                        }
                        cmd = splicCmd(bytes, content);
                        startConsumer(netSocket, cmd);
//                    }

                    break;
                case "01":
//                        88810103230000000000000000380C3230313831303236303030301438393836303431333039313833303039363432380D313061312C303030303963343318020104BB000844D31D0D0D0D0D
//                             8881010323000000000000000038
//                             0C323031383130323630303030
//                             1438393836303431333039313833303039363432380D313061312C303030303963343318020104BB000844D31D0D0D0D0D
                    cmd = splicCmd(bytes, "00");
                    startConsumer(netSocket, cmd);

                    int dtu_idLength = Utils.bytestoint(bytes, 14);
                    NumEntity dtu = getStringValue(14, bytes);
                    String dtuId = dtu.getValue();
                    NumEntity icc = getStringValue(dtu.getInit(), bytes);
                    String iccId = icc.getValue();
                    NumEntity cell = getStringValue(icc.getInit(), bytes);
                    String cellCode = cell.getValue();
                    start = cell.getInit();
                    String softVersion = Utils.bytes2String(bytes, start);
                    String hardVersion = Utils.bytes2String(bytes, start + 1);
                    int deviceCount = Utils.bytestoint(bytes, start + 2);
                    start = start + 3;
                    for (int i = 0; i < deviceCount; i++) {
                        deviceLenth = Utils.bytestoint(bytes, start);
                        deviceC = Utils.bytes2String(bytes, start + 1, deviceLenth);
                        deviceCode=deviceC;
                        deviceCodeList.add(deviceC);
                        start = start + 1 + deviceLenth;
                    }
                    logger.info("iccId:{}==>deviceCode:{}==>softVersion:{}==>hardVersion:{}",iccId,deviceCode,softVersion,hardVersion);
                    receiveCmd(queue, deviceCode);
                    break;
                case "02":
//                    88810203B80000000000000000011AB6E70D0D0D0D

                    cmd = splicCmd(bytes, "");
                    startConsumer(netSocket, cmd);

                    int sigal = Utils.bytestoint(bytes, start);
                    JsonObject query = new JsonObject().put("port", port).put("deviceId", deviceCode).put("type", 200);
                    JsonObject entity = query.copy();
                    entity.put("deviceType", "678");
                    entity.put("v", sigal);
                    entity.put("des", "信号强度");
                    entity.put("createdAt", new JsonObject().put("$date", currentTime));
                    mongoClient.bulkWrite(MONGONAME, Arrays.asList(BulkOperation.createInsert(entity)), stringAsyncResult -> {
                        if (stringAsyncResult.failed()) {
                            logger.error("monggoDB save faile:{}==>port:{}==>deviceCode{}==>HandId:{}" ,stringAsyncResult.cause().getMessage(),port ,deviceCode,handlerId);
                        }
                    });
//                    JsonObject update = new JsonObject().put("$set", entity);
//                    UpdateOptions options = new UpdateOptions().setMulti(false).setUpsert(true);
//                    mongoClient.updateCollectionWithOptions(MONGONAME_REAL, query, update, options, res -> {
//                        if (res.succeeded()) {
//                        } else {
//                            logger.error("deviceCode:{}==>update sigal error}",deviceCode);
//                        }
//                    });

                    logger.info("port: {}==>HandlerID:[{}]==>deviceCode:[{}]--heart only",port,handlerId,deviceCode);
                    break;
                case "03":
                    cmd = splicCmd(bytes, "00");
                    startConsumer(netSocket, cmd);

                    int signal = Utils.bytestoint(bytes, start);
                    int uploadCount = Utils.bytestoint(bytes, start + 1);
                    int deviceIdLength = Utils.bytestoint(bytes, start + 2);
                    String deviceId = Utils.bytes2String(bytes, start + 3, deviceIdLength);
                    if (deviceCode == null) {
                        deviceCode = deviceId;
                        deviceCodeList.add(deviceId);
                    }
                    start = start + 3 + deviceIdLength;
                    int isRing = Utils.bytestoint(bytes, start);
                    int deviceChannelCount = Utils.bytestoint(bytes, start + 1);
                    List<BulkOperation> bulkOperationList = new ArrayList<>();
                    JsonArray lasteDate = new JsonArray();
                    JsonArray policeDate = new JsonArray();
                    start = start + 2;
                    Map<String,Double> mapValue =new ConcurrentHashMap<>();
                    for (int i = 0; i < deviceChannelCount; i++) {
                        String deviceChannel = Utils.toBinary(bytes, start);
                        int deviceNum = Integer.parseInt(deviceChannel.substring(0, 4), 2);
                        int deviceType = Integer.parseInt(deviceChannel.substring(4), 2);
                        byte[] channelData = Utils.getbytes(bytes, start + 1);
                        int deviceStatus = Utils.toBinaryToint(channelData, 0, 4);
                        int dataLength = Utils.toBinaryToint(channelData, 4);
                        int dataValue = Utils.bytestoint(bytes, start + 2, dataLength);
                        String unit = null;
                        BigDecimal bigDecimal = new BigDecimal(dataValue);
                        int type = 0;
                        String dec=null;
                        Map<String, Integer> ele = electricParmas.get(deviceId);
                        if (deviceType == 1) {
                            bigDecimal.setScale(1);
                            bigDecimal = bigDecimal.divide(new BigDecimal(10));
                            unit = "℃";
                            type = deviceNum + 4;
                            if(deviceNum==1){
                                dec="第一路温度";
                            }else if (deviceNum==2){
                                dec="第二路温度";
                            }else if (deviceNum==3){
                                dec="第三路温度";
                            }else if (deviceNum==4){
                                dec="第四路温度";
                            }
                        } else if (deviceType == 2) {
                            unit = "mA";
                            type = 63;
                            dec="剩余电流";
                        } else if (deviceType == 3) {
                            int ct;
                            if(ele==null){
                                ct=1;
                            }else {
                                ct = ele.get("ct");
                            }
                            bigDecimal.setScale(2);
                            bigDecimal = bigDecimal.divide(new BigDecimal(100)).multiply(new BigDecimal(ct));
                            unit = "A";
                            type = deviceNum + 11;
                            double v = bigDecimal.doubleValue();
                            if(deviceNum==1){
                                dec="A相电流";
                                mapValue.put("AVoltage", v);
                            }else if (deviceNum==2){
                                dec="B相电流";
                                mapValue.put("BVoltage", v);
                            }else if (deviceNum==3){
                                dec="C相电流";
                                mapValue.put("CVoltage", v);
                            }
                        } else if (deviceType == 5) {
                            int pt;
                            if(ele==null){
                                pt=1;
                            }else {
                                pt = ele.get("pt");
                            }
                            bigDecimal.setScale(1);
                            bigDecimal = bigDecimal.divide(new BigDecimal(10)).multiply(new BigDecimal(pt));
                            unit = "V";
                            type = deviceNum + 8;
                            double v = bigDecimal.doubleValue();
                            if(deviceNum==1){
                                dec="A相电压";
                                mapValue.put("A", v);
                            }else if (deviceNum==2){
                                dec="B相电压";
                                mapValue.put("B", v);
                            }else if (deviceNum==3){
                                dec="C相电压";
                                mapValue.put("C", v);
                            }
                        }
                        double v = bigDecimal.doubleValue();
                        start = start + 2 + dataLength;
                        JsonObject jsonDate = new JsonObject();
                        jsonDate.put("deviceId", deviceId);
                        jsonDate.put("port", port);
                        jsonDate.put("deviceNum", deviceNum);
                        jsonDate.put("deviceType", deviceType);
                        jsonDate.put("deviceStatus", deviceStatus);
                        jsonDate.put("value", v);
                        jsonDate.put("type", type);
                        jsonDate.put("unit", unit);
                        jsonDate.put("des", dec);
                        JsonObject laste = jsonDate.copy();
                        lasteDate.add(laste);
                        jsonDate.put("createdAt", new JsonObject().put("$date", currentTime));
                        laste.put("createdTime", currentTime);
                        bulkOperationList.add(BulkOperation.createInsert(jsonDate));
                        if(deviceType<3&&deviceStatus==1){
                            policeDate.add(laste);
                        }
                    }

                    int fauCount = Utils.bytestoint(bytes, start);
                    start = start + 1;
                    for (int i = 0; i < fauCount; i++) {
                        JsonObject laste = new JsonObject();

                        int sta = Utils.bytestoint(bytes, start);
                        String fauType = Utils.bytes2String(bytes, start+=1, sta);
                        start = start + sta;
                        int type = 0;
                        switch (fauType) {
                            case "501F":
                                type = 91;
                                break;
                            case "502F":
                                type = 92;
                                break;
                            case "501A":
                                type = 91;
                                break;
                            case "501B":
                                type = 101;
                                break;
                            case "501C":
                                type = 111;
                                break;
                            case "502A":
                                type = 92;
                                break;
                            case "502B":
                                type = 102;
                                break;
                            case "502C":
                                type = 112;
                                break;
                            case "503F":
                                type = 150;
                                break;
                            case "504F":
                                type = 159;
                                break;
                            case "301F":
                                type = 151;
                                break;
                            case "301A":
                                type = 152;
                                break;
                            case "301B":
                                type = 153;
                                break;
                            case "301C":
                                type = 154;
                                break;
                            case "302F":
                                type = 155;
                                break;
                            case "302A":
                                type = 156;
                                break;
                            case "302B":
                                type = 157;
                                break;
                            case "302C":
                                type = 158;
                                break;
                            default:
                                type = Integer.parseInt(fauType.substring(0, fauType.length() - 1));
                        }
                        if(fauType.startsWith("50")&&!StringUtils.equals("504F", fauType)&&!StringUtils.equals("503F", fauType)) {
                            if (fauType.endsWith("F")){
                                continue;
                            }
                            double va = 0;
                            if (fauType.endsWith("A")) {
                                va = mapValue.get("A");
                            } else if (fauType.endsWith("B")) {
                                va = mapValue.get("B");
                            } else if (fauType.endsWith("C")) {
                                va = mapValue.get("C");
                            }
                            laste.put("value", va);
                            laste.put("unit", "V");
                        }else  if(fauType.startsWith("30")) {
                            if (fauType.endsWith("F")){
                                continue;
                            }
                            double va = 0;
                            if (fauType.endsWith("A")) {
                                va = mapValue.get("AVoltage");
                            } else if (fauType.endsWith("B")) {
                                va = mapValue.get("BVoltage");
                            } else if (fauType.endsWith("C")) {
                                va = mapValue.get("CVoltage");
                            }
                            laste.put("value", va);
                            laste.put("unit", "A");
                        }
                        laste.put("deviceId", deviceId);
                        laste.put("port", port);
                        laste.put("deviceType", deviceType);
                        laste.put("type", type);
                        laste.put("createdTime", currentTime);
                        policeDate.add(laste);
                    }

                    upLine( deviceCode, keepLiveMap,handlerId);
                    if (!policeDate.isEmpty()) {
                        eventBus.send(Constant.MOD_DEVICE_POLICE, returncurrResult(deviceCode,policeDate));
                    }

                    if (!bulkOperationList.isEmpty()) {
                        saveMongo("RealTimeData", mongoClient, bulkOperationList, eventBus,lasteDate);
                    }

                    break;
                case "A1":
                case "B0":
                    CommandQueueEle ele = lookReturn(flag);
                    int suc = Utils.bytestoint(bytes, start);
                    boolean sucess = false;
                    if (suc == 0) {
                        sucess = true;
                    }
                    if (ele!=null){
                        eventBus.send(Constant.MOD_CALLBACK, returncurrResult(deviceCode, JsonObject.mapFrom(ele).put("success", sucess)));
                    }
                    break;
                case "A0":
                    //8881A0030A0000000000000000A7
                    // 0021
                    // 2101020001
                    // 21020207D0
                    // 21030200011101020001110202004611030200641201020001120202004612030200641301020001130202004613030200641401020001140202004614030200640001020001000202001E501102000050120200F25013020064502102000050220200BB502302006450310200015033020064504102000050430200643011020000301202000A30130200643021020000302202000A302302006462C40D0D0D0D
                    //8881A0030A0000000000000000A7
                    // 00212101020001210202012C21030200011101020001110202004611030200641201020001120202004612030200641301020001130202004613030200641401020001140202004614030200640001020001000202001E501102000050120200F25013020064502102000050220200BB502302006450310200015033020064504102000050430200643011020000301202000A30130200643021020000302202000A3023020064EC570D0D0D0D
                    ele = lookReturn(flag);
                    int paramsCount = Utils.bytestoint(bytes, start, 2);
                    start = start + 2;
                    JsonArray calldata = new JsonArray();
                    for (int i = 0; i < paramsCount; i++) {
                        JsonObject entries = new JsonObject();
                        String type = Utils.bytes2String(bytes, start, 2);
                        int len=Utils.bytestoint(bytes,start+=2);
                        int value = Utils.bytestoint(bytes, start += 1, len);
                        start+=len;
                        entries.put("type", type);
                        entries.put("value", value);
                        calldata.add(entries);
                    }

                    if (ele!=null){
                        eventBus.send(Constant.MOD_CALLBACK, returncurrResult(deviceCode, JsonObject.mapFrom(ele).put("success", true).put("calldata", calldata)));
                    }
                    break;

                default:
                    System.out.println("aksdjskjhfksdhkfjhsdkfsdkjh-------------------------------------------");
            }
            return true;
        }

        public CommandQueueEle lookReturn(String flag){
            CommandQueueEle ele;
            do {
                ele = returnQueue.poll();
                if (ele == null) {
                    logger.info("queue not message  atuo send ");
                    break;
                }
                String value = ele.getValue();

                logger.debug("value: " +value+"    resultPK:"+flag);
                if (StringUtils.equals(value, flag)) {
                    break;
                }  else{
                    if (ele.isCallback()){
                        eventBus.send(Constant.MOD_CALLBACK, returncurrResult(deviceCode,JsonObject.mapFrom(ele).put("success", false).put("status", 3)));
                    }
                    ele=null;
                    logger.info("dump message cmd: " +ele.getCmd());
                }

            } while (returnQueue.size() > 0);
            return ele;
        }
        public void setTimer(NetSocket netSocket ) {
            vertx.setTimer(1000, h->{
                if(deviceCode!=null){
                    eventBus.localConsumer(Constant.ONLY_CMD + port+"_"+deviceCode,obj ->{
                        JsonObject json = (JsonObject)obj.body();
                        queue.offer(new Message(json.getString("cmd"), 4, json.getString("source"), json.getString("messageId"), true,deviceCode));

//                        startConsumer(netSocket,json.getString("cmd"));
                    });
                }else {
                    setTimer(netSocket);
                }
            });

        }

        private void saveMongo(String dataName,MongoClient mongoClient,List<BulkOperation> bulkOperationList,EventBus eventBus,JsonArray lasteDate) {
            mongoClient.bulkWrite(dataName, bulkOperationList, stringAsyncResult -> {
                if (stringAsyncResult.succeeded()) {
                    eventBus.send(Constant.MOD_LATEST, returncurrResult(deviceCode, lasteDate.toString()));
                } else {
                    logger.error("monggoDB save faile： " + stringAsyncResult.cause().getMessage() + "==>port:" + port + "==>" + deviceCode);
                }
            });
//		bulkOperationList.forEach((a) -> {
//			logger.info("bulkOperationList： " + a.getDocument().toString());
//		});
        }
    }

    @Override
    public void stop() {
        logger.info("Collector client ModbusModuleServer undeployed!!!" + "==>port:" + port + "==>");
    }

    private void startConsumer(NetSocket netSocket, List<Long> timerOther, PriorityBlockingQueue<Message> queue, ConcurrentLinkedQueue<CommandQueueEle> returnQueue,Map<String, Integer> lineStatusMap
   ,List<String> deviceCodeList ) {
        long periodicId = vertx.setPeriodic(1000, id -> {
            int queueSize = queue.size();
            Message peek = queue.peek();
            if (queue.peek() != null) {
                String devicePk=peek.getDevicePk();
                boolean lineStatus = lineStatusMap.get(devicePk)>0;
                if (!lineStatus) {
                    logger.info("port:{}==>deviceCode:{}==>device not conection, not send cmd ,queue not consumer count {} cmd:{}" ,port,devicePk,queueSize,peek.getCmd());
                    if (queueSize > 15) {
                        queue.clear();
                    }
                    return;
                }
                if (netSocket.writeQueueFull() && queueSize > 0) {
                    logger.info("情况不对,赶快撤退 only==>port:{}==>deviceCode:{}==>{}==>队列未消费数:{}" ,port,devicePk,netSocket.writeHandlerID(),queueSize);
                    if (queueSize > 15) {
                        queue.clear();
                    }
                    return;
                }
                Message receive = queue.poll();
                String cmdStr = receive.getCmd();
                Buffer buffer = Buffer.buffer().appendBytes(Utils.hexStringToByteArray(cmdStr));
                netSocket.write(buffer, result -> {
                    if (result.failed()) {
                        logger.debug(port + "cmd send fail-info:" + result.cause().getMessage());
                    }else {
                        returnQueue.offer(new CommandQueueEle(System.currentTimeMillis(), cmdStr, true, cmdStr.substring(4, 6),receive.getSource(), receive.getMessageId()));
                    }
                });

                logger.info(netSocket.writeHandlerID() + "==>队列个数:{}==>port:{}==>deviceCode:{}==>cmd:" ,queue.size(),port,devicePk,cmdStr);
            }
        });

        long returnPeriodicId = vertx.setPeriodic(1000*2, id -> {
            CommandQueueEle peek = returnQueue.peek();
            if (peek != null) {
                long l = System.currentTimeMillis();
                long time = peek.getTime();
                if (l - time > 3000) {
                    if (!deviceCodeList.isEmpty()) {
                        returnQueue.poll();
                        eventBus.send(Constant.MOD_CALLBACK, returncurrResult(deviceCodeList.get(0), JsonObject.mapFrom(peek).put("success", false)));
                    }
                }
            }
        });
        timerOther.add(periodicId);
        timerOther.add(returnPeriodicId);


    }

    private void checkLine(long time, long timeout, Map<String, Long> keepLiveMap, NetSocket netSocket,  List<Long> timerOther,List<String> deviceCodeList) {
        long overtimeId = vertx.setPeriodic(time, t -> {
            if (keepLiveMap.get("alive") != null
                    && (System.currentTimeMillis() - keepLiveMap.get("alive") > timeout) && !netSocket.writeQueueFull()) {
                timeOutClose(netSocket, keepLiveMap,deviceCodeList);
            }
        });
        timerOther.add(overtimeId);
    }

    public void timeOutClose(NetSocket netSocket, Map<String, Long> keepLiveMap,List<String> deviceCodeLis) {
        String nowTime = TimeUtil.parseTime(TimeUtil.getDateTimeOfTimestamp(System.currentTimeMillis()));
        String startTime = TimeUtil.parseTime(TimeUtil.getDateTimeOfTimestamp(keepLiveMap.get("alive")));
        if (!deviceCodeLis.isEmpty()){
            logger.info( "HandlerID: {}----alive: only contection timeout==>port:{}==>deviceCode:{}===>nowTime:{} ==>startTime:{}" ,netSocket.writeHandlerID(),port,deviceCodeLis.get(0),nowTime,startTime);
        }else {
            logger.info( "HandlerID: {}----alive: only contection timeout==>port:{}==>nowTime:{} ==>startTime:{}" ,netSocket.writeHandlerID(),port,nowTime,startTime);
        }
        netSocket.close(h->{
            logger.debug("connect close status end only:{}==>port:{}==>HandlerID:{}==>",h.succeeded(),port,netSocket.writeHandlerID());
            if(h.failed()){
                logger.debug("connect close status end only:{}==>port:{}==>HandlerID:{}==>result",h.succeeded(),port,netSocket.writeHandlerID(),h.cause().getMessage());
            }
        });
    }

    public void calcCount(SharedData sharedData, boolean isAdd) {
        sharedData.getCounter("mycounter", res -> {
            if (res.succeeded()) {
                Counter counter = res.result();
                if (isAdd) {
                    counter.incrementAndGet(count -> {
                        if (count.succeeded()) {
                            logger.info("only add after current total run count: {}",count.result()  );
                        }
                    });
                } else {
                    counter.decrementAndGet(count -> {
                        if (count.succeeded()) {
                            logger.info(" only decrement after current total run count: {}",count.result() );
                        }
                    });
                }

            }
        });
    }

    private void cancelTimer(List<Long> map){
        map.forEach(e -> {
            vertx.cancelTimer(e);
        });
    }

    public void downLine(List<String> deviceCodeList,String handlerId) {
        deviceCodeList.forEach( deviceCode ->{
            downLine( deviceCode,handlerId);
        });
    }

    public void downLine(String deviceCode,String handlerId) {
        Integer lineCount = lineStatusMap.get(deviceCode);
        lineCount = lineCount - 1;
        if (handlerIdMap.containsKey(handlerId)){
            handlerIdMap.remove(handlerId);
        }
        lineStatusMap.put(deviceCode, lineCount);
        logger.info("downLine==>port:{}==>deviceCode:{}==>HandlerId:{}==>lineCount:{}",port,deviceCode,handlerId ,lineCount);
        if (lineCount < 1) {
            eventBus.send(Constant.MOD_OFFLINE_STATUS, returncurrResult(deviceCode, new JsonObject().put("status", 0)));
        }
    }

    public void upLine(List<String> deviceCodeList,Map<String, Long> keepLiveMap,String handlerId ) {
        deviceCodeList.forEach( deviceCode ->{
            upLine(deviceCode,keepLiveMap,handlerId);
        });
    }

    public void upLine(String deviceCode,Map<String, Long> keepLiveMap,String handlerId ) {
        if(!lineStatusMap.containsKey(deviceCode)){
            logger.error("no exis deviceCode:"+deviceCode);
            return;
        }
        Integer lineCount = lineStatusMap.get(deviceCode);
        boolean lineStatus =lineCount>0;
        if (!lineStatus) {
            eventBus.send(Constant.MOD_ONLINE_STATUS
                    , returncurrResult(deviceCode,new JsonObject().put("status", 1)));
        }
        logger.info("upline==>port:{}==>deviceCode:{}==>HandlerId:{}==>lineCount:{}",port,deviceCode,handlerId ,lineCount);
        if (!handlerIdMap.containsKey(handlerId)){
            lineStatusMap.put(deviceCode,lineCount+1);
            handlerIdMap.put(handlerId, 1);
        }
        keepLiveMap.put("alive", System.currentTimeMillis());
    }

    public  void receiveCmd(PriorityBlockingQueue<Message> queue ,String deviceCode) {
            eventBus.localConsumer("send.cmd." + port + deviceCode, obj -> {
                boolean lineStatus = lineStatusMap.get(deviceCode)>0;
                JsonObject json = (JsonObject) obj.body();
                Message message = json.mapTo(Message.class);
                if (!lineStatus) {
                    eventBus.send(Constant.MOD_CALLBACK
                            , returncurrResult(deviceCode, new JsonObject().put("status", 0).put("success", false).put("messageId", json.getString("messageId"))));
                } else {
                    logger.info(json.getString("cmd"));
                    queue.offer(message);
                }
            });

    }


    public JsonObject returncurrResult(String deviceCode,Object returnData){
        return new JsonObject().put("dtuType", dtuType).put("port", port).put("deviceType", deviceType).put("createTime",TimeUtil.getCurrentDatetime()).put("deviceCode", deviceCode).put("data", returnData);
    }

    private void startConsumer(NetSocket netSocket,String cmd) {
        Buffer buffer1 = Buffer.buffer().appendBytes(Utils.hexStringToByteArray(cmd));
        if (netSocket.writeQueueFull()) {
            logger.info("情况不对,赶快撤退==>port:{}==>deviceCode:{}==>HandlerID:{}", port, netSocket.writeHandlerID());
        }
        netSocket.write(buffer1, result -> {
            if (result.failed()) {
                logger.debug("port:{}cmd send fail-info:" + cmd + result.cause().getMessage(), port);
            } else {
                logger.debug("port:{} send success-info cmd:{},HandlerID:{}", port, cmd, netSocket.writeHandlerID());
            }
        });
    }
    public String splicCmd(byte[] bytes,String data){
        String start = Utils.bytes2String(bytes,0,13);
       return splicCmd(start,data);
    }

    public static String splicCmd(String start,String data){
        String s = Utils.addStr(start, String.format("%02x",data.length()/2),data);
        String cmd = Utils.checkNumSecondAll(s) + "0D0D0D0D";
        return cmd.toUpperCase();
    }

    public static String splicCmd(String start,String flag,String midd,String data){
        String s = Utils.addStr(start,flag,midd, String.format("%02x",data.length()/2),data);
        String cmd = Utils.checkNumSecondAll(s) + "0D0D0D0D";
        return cmd.toUpperCase();
    }


    public NumEntity getStringValue(Integer initValue,byte[] bytes){
        int iddIdLength = Utils.bytestoint(bytes, initValue);
        byte[] iddIdBytes = Utils.getbytes(bytes, initValue+1, iddIdLength);
        String iccid = new String(iddIdBytes);
        return  new NumEntity(initValue+1+iddIdLength,iccid);
    }

    class NumEntity {
        @Getter
        @Setter
        private int init;
        @Getter
        @Setter
        private String value;

        public NumEntity(int init, String value) {
            this.init = init;
            this.value = value;
        }
    }

   public class Ratio {
        @Getter
        @Setter
        private int pt;
        @Getter
        @Setter
        private int ct;

       public Ratio(int pt, int ct) {
           this.pt = pt;
           this.ct = ct;
       }
   }

}
 